package org.atlantis.datasink.event;

import static org.atlantis.datasink.event.EventQueueContext.EXITED;
import static org.atlantis.datasink.event.EventQueueContext.RUNNING;
import static org.atlantis.datasink.event.EventQueueContext.WAITING;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicInteger;

import org.atlantis.datasink.mx.EventMonitor;
import org.atlantis.datasink.utils.JMXHelper;
import org.atlantis.datasink.utils.ToStringHelper;
import org.atlantis.datasink.utils.jsr166y.LinkedTransferQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

final class LimitedEventQueue implements EventQueue {

	private static final Logger log = LoggerFactory.getLogger(EventQueue.class);

	private String name;
	private Queue<Object> messageQueue = new LinkedTransferQueue<Object>();
	private Queue<EventHandler> handlerQueue = new LinkedTransferQueue<EventHandler>();
	private int messageLimit;
	private int workerLimit;
	private AtomicInteger messageDefender = new AtomicInteger(0);
	private AtomicInteger workerDefender = new AtomicInteger(0);

	private EventQueueContext ctx;

	private EventMonitor monitor;

	LimitedEventQueue(String name, int messageLimit, int workerLimit) {
		this.name = name;
		this.messageLimit = messageLimit;
		this.workerLimit = workerLimit;
		this.ctx = new EventQueueContext();
		this.monitor = new EventMonitor(name, messageLimit, workerLimit);
		JMXHelper.registerMBean(this.monitor);
	}

	void execute(Object message) {
		monitor.setCurrentMessageSize(messageDefender.get());
		Iterator<EventHandler> handlerIter = handlerQueue.iterator();
		while (handlerIter.hasNext()) {
			EventHandler handler = handlerIter.next();
			handler.onEvent(message);
		}
		monitor.incTotalMessageSize();
	}

	@Override
	public String getName() {
		return name;
	}

	@Override
	public boolean offer(Object message) {
		int status = ctx.status.get();
		if (status == EXITED) return false;
		if (status == WAITING) {
			if (ctx.status.compareAndSet(WAITING, RUNNING)) {
				workerDefender.incrementAndGet();
				EventManager.getInstance().execute(new MessageWorker());
			}
		}

		if (messageDefender.incrementAndGet() <= messageLimit) {
			return messageQueue.offer(message);
		}
		messageDefender.decrementAndGet();

		if (workerDefender.incrementAndGet() <= workerLimit) {
			EventManager.getInstance().execute(new MessageWorker(message));
			return true;
		}
		workerDefender.decrementAndGet();

		monitor.incDiscardMessageSize();
		log.warn("No more resource for " + toString());
		return false;
	}

	@Override
	public int size() {
		return messageDefender.get();
	}

	@Override
	public boolean addHandler(EventHandler handler) {
		if (!isAlive()) return false;
		if (handlerQueue.contains(handler)) return false;
		return handlerQueue.offer(handler);
	}

	@Override
	public boolean removeHandler(EventHandler handler) {
		if (!isAlive()) return false;
		return handlerQueue.remove(handler);
	}

	@Override
	public Collection<EventHandler> getHandlers() {
		List<EventHandler> immutable = new ArrayList<EventHandler>(handlerQueue);
		return immutable;
	}

	@Override
	public Collection<Object> stop() {
		ctx.status.set(EXITED);

		JMXHelper.unregisterMBean(monitor);

		Collection<Object> remainingMessage = new ArrayList<Object>(messageQueue);
		messageQueue.clear();
		handlerQueue.clear();
		// help GC
		messageQueue = null;
		handlerQueue = null;
		return remainingMessage;
	}

	@Override
	public boolean isAlive() {
		return ctx.status.get() != EXITED;
	}

	@Override
	public String toString() {
		ToStringHelper helper = ToStringHelper.create(EventQueue.class);
		helper.add("name", name).add("messageLimit", messageLimit).add("currentMessageSize", messageDefender.get())
				.add("workerLimit", workerLimit).add("currentWorkerSize", workerDefender.get())
				.add("handlerSize", handlerQueue.size());
		return helper.end();
	}

	private final class MessageWorker implements Runnable {
		private Object initMessage;

		MessageWorker() {}

		MessageWorker(Object initMessage) {
			this.initMessage = initMessage;
		}

		@Override
		public void run() {
			monitor.setCurrentWorkerSize(workerDefender.get());
			EventNamedThreadFactory.modifyName(name);
			if (initMessage != null) execute(initMessage);
			Object message = null;
			while ((message = messageQueue.poll()) != null) {
				messageDefender.decrementAndGet();
				execute(message);
				if (ctx.status.get() == EXITED) break;
			}
			if (workerDefender.decrementAndGet() <= 0) ctx.status.compareAndSet(RUNNING, WAITING);
			EventNamedThreadFactory.restoreName();
			monitor.setCurrentWorkerSize(workerDefender.get());
		}
	}
}
